package bll

import (
	"bytes"
	"context"
	"fmt"
	"math"
	"math/rand"
	"signal/event"
	"signal/lib/trace_log"
	"signal/utils/goroutine_pool"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"gitlab.local/backend/proto/dto"
	"signal/client/grpc"
	log "signal/collector/logger"
	"signal/errors"
	"signal/model"
	"signal/utils"

	"github.com/panjf2000/gnet/v2"
	"go.uber.org/zap"
)

const (
	authDataLength      = 17 // 登录认证的数据长度
	snCodeLength        = 16
	requestRetryCount   = 1
	replyHeartBeatCount = 3
)

var (
	randSource    = rand.NewSource(time.Now().UnixNano())
	randGenerator = rand.New(randSource)
)
var (
	sessionMissingErr = errors.New("session not found")
	connMissingErr    = errors.New("connection not found")
	noLoginErr        = errors.New("device is not logged in")
	offlineErr        = errors.New("device is offline")
	badLoginErr       = errors.New("bad login data")
	badLoginStepErr   = errors.New("bad login step")
	badRemoteAddrErr  = errors.New("bad remote addr")
	noReplyErr        = errors.New("got no reply after retries")
	heartbeatErr      = errors.New("too many retries to reply heart beat")
)

type closeData struct {
	Sn     string
	ConnId int
	Ch     chan string
}

// RequestKey 生成session.requests的key
func RequestKey(sn string, cmd model.CmdType, requestId uint8) string {
	return fmt.Sprintf("%s-%d-%d", sn, cmd, requestId)
}

var Session = &session{}

type session struct {

	// key: connId
	// value: []byte
	// cachePacket map[int][]byte
	cachePacket *sync.Map

	// key: connId
	// value: connection status
	// status map[string]*Status
	status *sync.Map

	// key: sn
	// value: tcp connection
	// conns map[string]gnet.Conn
	conns *sync.Map

	// record manager request to the device
	// key: sn+cmd+id
	// value: response data
	// requests map[string]chan interface{}
	requests *sync.Map

	// key: sn+cmd+id
	// value: retry count
	// requestRetryMap map[string]int
	requestRetryMap *sync.Map

	CloseMap *sync.Map

	CloseChan chan *closeData

	// 人数
	connectCount int32
}

// Status save device session state
// nonce is time format as YYMMDDHHMMSS
// keepalive record device connection timeout
type status struct {
	sn        []byte
	nonce     []byte // YYMMDDHHMMSS 格式的时间
	login     bool   // 是否已登录
	keepalive int64  // 连接超时时间，单位秒，若通信间隔超过该时间则断开连接
	aesKey    []byte
	iv        []byte
	//idCreator *utils.ID
	IsExpire bool // 连接是否过期，即下线
}

//func (r *status) getID() uint8 {
//	return r.idCreator.Get()
//}
//
//func (r *status) getIDUint64() uint64 {
//	return r.idCreator.GetUInt64()
//}
//
//func (r *status) idLen() int {
//	return r.idCreator.Len()
//}
//
//func (r *status) release(id uint8) {
//	r.idCreator.Release(id)
//}

const (
	loginRequest          = 0x01 // 【上行】设备申请登录
	responseEncryptedData = 0x02 // 【下行】服务端向设备返回加密数据
	gotEncryptedData      = 0x03 // 【上行】设备加密验证数据给服务端
	loginSuccess          = 0x04 // 【下行】服务端向设备告知登录成功
)

func (l *session) Init() func() {
	return l.init()
}

func (l *session) init() func() {
	l.cachePacket = &sync.Map{}
	l.status = &sync.Map{}
	l.conns = &sync.Map{}
	l.requests = &sync.Map{}
	l.requestRetryMap = &sync.Map{}
	ch := make(chan struct{})
	l.CloseMap = &sync.Map{}
	l.CloseChan = make(chan *closeData, 10)
	go l.checkConnections(ch)
	go l.checkCloseChan(l.CloseChan)
	return func() {
		close(ch)
	}
}

func (l *session) GetRequestID(sn string, cmd model.CmdType) (string, uint8, error) {

	_, err := l.getStatus(sn)
	if err != nil {
		return "", 0, err
	}

	/*
	   从一个随机数开始+1递增,如果1-255个数都遍历完了都有重复了，那么就返回初始的id
	*/
	var (
		startId = uint8(randGenerator.Intn(math.MaxUint8) + 1)
	)
	for {
		var (
			id         = startId
			requestKey = RequestKey(sn, cmd, id)
		)

		if _, ok := l.requests.Load(requestKey); ok {
			id = id%math.MaxUint8 + 1
			if id == startId {
				return RequestKey(sn, cmd, id), id, nil
			}
			continue
		}
		return requestKey, id, nil
	}
}

func (l *session) getStatus(sn string) (*status, error) {
	value, ok := l.conns.Load(sn)
	if !ok || value == nil {
		// 设备下线
		_ = Device.Offline(sn)

		// 设备离线
		eData := &dto.DeviceStatus{
			Sn:    sn,
			Event: dto.EventType_Offline,
		}
		event.E.PublishSync(event.DeviceStatusChange, eData)

		// 没有信号
		ed := &dto.Signal{
			Sn:        sn,
			SignalGps: int32(0),
			Signal_4G: int32(0),
		}

		event.E.PublishSync(event.DeviceSignalReport, ed)

		return nil, offlineErr
	}
	c := value.(gnet.Conn)

	v, ok := l.status.Load(c.Fd())
	if !ok || v == nil {
		return nil, offlineErr
	}

	s := v.(*status)
	return s, nil
}

func (l *session) ReleaseID(sn string, id uint8) error {
	value, ok := l.conns.Load(sn)
	if !ok || value == nil {
		return connMissingErr
	}
	c := value.(gnet.Conn)
	if c.RemoteAddr() == nil {
		return badRemoteAddrErr
	}

	v, ok := l.status.Load(c.Fd())
	if !ok || v == nil {
		return sessionMissingErr
	}

	//fmt.Println("----------------- connection release id: ", id, " sn: ", sn)
	//s := v.(*status)
	//
	//s.release(id)
	return nil
}

func (l *session) GetSnByConnId(connId int) string {
	if ret, ok := l.status.Load(connId); ok && ret != nil {
		s := ret.(*status)
		return string(s.sn)
	}
	return ""
}

func (l *session) UpdateKeepalive(connId int, keepalive int64) {
	result, ok := l.status.Load(connId)
	if ok && result != nil {
		status := result.(*status)
		status.keepalive = keepalive
		sn := l.GetSnByConnId(connId)
		if sn != "" {
			trace_log.DebugLogger(sn, "update keepalive: ", keepalive, " expireTime:", utils.UnixToFormat(keepalive), "connId: ", connId, "sn: ", sn)
		}
	}
}

func (l *session) IsLogged(connId int) bool {
	result, ok := l.status.Load(connId)
	if ok && result != nil {
		s := result.(*status)
		return s.login
	}
	return false
}

func (l *session) Login(c gnet.Conn, data []byte) ([]byte, error) {

	if len(data) != authDataLength {
		return nil, badLoginErr
	}

	var (
		snCodeStart = 1
		snCodeEnd   = 1 + snCodeLength
		step        = data[0]
		err         error
		snBytes     = data[snCodeStart:snCodeEnd]
		response    []byte
	)

	fmt.Printf("---------- session login[%v] sn: %v, 0x%X\n", step, string(snBytes), snBytes)
	switch step {
	case loginRequest:
		response, err = l.handleAuthRequest(c, snBytes)
	case gotEncryptedData:
		response, err = l.handleVerifyRequest(c, snBytes)
	default:
		return nil, badLoginStepErr
	}
	if err != nil {
		return nil, err
	}

	return response, err
}

// 处理设备登录认证请求
func (l *session) handleAuthRequest(c gnet.Conn, data []byte) ([]byte, error) {
	var (
		connId    = c.Fd()
		err       error
		key       = string(data)
		pkt       *model.Packet
		aes       *dto.AesInfoReplay
		body      []byte
		encrypted []byte
		ctx, _    = context.WithTimeout(context.Background(), time.Minute*3)
	)

	trace_log.DebugLogger(string(data), "开始处理登录认证请求 step 2  success connId:", connId)
	if aes, err = grpc.Device.AesInfo(ctx, &dto.AesInfoRequest{Sn: string(data)}); err != nil {
		log.Errorf("get aes info error: %+v", err)
		return nil, err
	}
	log.Infof("got aes_key: %s", aes)

	// 1. get and parse sn_code
	nonce := utils.GetTimestampHex()
	log.Infof("nonce : %X", nonce)
	log.Infof("sn : %X", data)
	s := &status{
		sn:        data,
		nonce:     nonce,
		keepalive: time.Now().Unix() + model.DefaultTCPTimeout,
		login:     false,
		aesKey:    aes.AesKey,
		iv:        aes.Iv,
		//idCreator: utils.NewID(),
		IsExpire: false,
	}

	// 2. encrypt by aes
	repeat := bytes.Repeat([]byte{0}, 16-len(nonce))
	nonce = append(nonce, repeat...)
	if encrypted, err = utils.EncryptAesWithCbc(nonce, aes.AesKey, aes.Iv); err != nil {
		return nil, err
	}

	// 旧连接缓存在
	if conn, ok := l.conns.Load(key); ok {
		trace_log.DebugLogger(string(data), " 旧连接已存在，断开旧连接")
		l.Disconnect(conn.(gnet.Conn), string(data), func(sn string) error {
			trace_log.DebugLogger(sn, "旧连接断开成功")
			return nil
		})

		trace_log.DebugLogger(string(data), "开始建立新连接")
	}

	// 3. save login status
	l.status.Store(connId, s)

	// 4. save connection
	l.conns.Store(key, c)

	// 5. respond time to device
	resp := &model.LoginResponse{
		Step: responseEncryptedData,
		Body: encrypted,
	}
	body = resp.ToHexBytes()
	pkt = model.BuildPacket(model.CmdLogin, 0, body)

	trace_log.DebugLogger(string(data), "处理登录请求完成 step 2  success connId:", connId)
	return pkt.ToHexBytes(), nil
}

// 验证设备用sn和服务端的时间加密的数据
func (l *session) handleVerifyRequest(c gnet.Conn, data []byte) ([]byte, error) {
	var (
		ok      bool
		err     error
		pkt     *model.Packet
		result  any
		content []byte
	)

	if result, ok = l.status.Load(c.Fd()); !ok || result == nil {
		return nil, sessionMissingErr
	}
	s := result.(*status)

	// 1. decode aes data
	if content, err = utils.DecryptAesWithCbc(data, s.aesKey, s.iv); err != nil {
		return nil, err
	}

	// 2. check data
	if !bytes.Equal(content[:4], s.nonce) || !bytes.Equal(content[8:], s.sn[8:]) {
		return nil, errors.Newf("authentication failed, ip connId: %v", c.Fd())
	}

	// 3. respond device session success or failed
	pkt = model.BuildPacket(model.CmdLogin, 0, []byte{loginSuccess})

	// 4. update device login status
	s.login = true
	l.status.Store(c.Fd(), s)

	// 设备上线，记录状态
	_ = Device.Online(string(s.sn))
	// emit event to other services
	eData := &dto.DeviceStatus{
		Sn:    string(s.sn),
		Event: dto.EventType_Online,
	}
	event.E.PublishSync(event.DeviceStatusChange, eData)
	return pkt.ToHexBytes(), err
}

// Send 向设备发送数据包
func (l *session) Send(sn string, data []byte) error {
	var (
		ok     bool
		err    error
		result any
		conn   gnet.Conn
	)

	if result, ok = l.conns.Load(sn); !ok || result == nil {
		log.Errorf("[%s] connection not found", sn)
		return connMissingErr
	}

	conn = result.(gnet.Conn)
	if !l.IsLogged(conn.Fd()) {
		return noLoginErr
	}

	if _, err = conn.Write(data); err != nil {
		return err
	}

	log.Infof("【Send】 data to device, data: %X", data)
	trace_log.DebugLogger(sn, "【Send】 data to device, data: ", fmt.Sprintf("%X", data), " connId:", conn.Fd())
	return conn.Flush()
}

func (l *session) TestRequest(sn, key string, data []byte) (interface{}, error) {
	var (
		err error
		ch  = make(chan interface{})
	)

	log.Infof("request %v %v", zap.String("sn", sn), zap.String("key", key))
	l.requests.Store(key, ch)
	defer func() {
		// delete request
		l.requests.Delete(key)
	}()

	_, ok := l.requests.Load(key)
	if ok {
		log.Info("store request ok", zap.String("key", key))
	}

	if err = l.Send(sn, data); err != nil {
		log.Errorf("send data to device error %s", err)
		return nil, err
	}

	trace_log.DebugLogger(sn, "BinaryReq send data to device success", string(data))
	// retry count
	retryCount, ok := l.requestRetryMap.Load(key)
	if ok {
		retryCount = retryCount.(int) + 1
	} else {
		retryCount = 1
	}

	l.requestRetryMap.Store(key, retryCount)

	log.Infof("wait for device [%s] response", sn)
	timer := time.NewTimer(time.Second * model.DefaultRequestTimeout)
	select {
	case <-timer.C:
		if retryCount.(int) >= requestRetryCount {
			defer func() {
				l.requestRetryMap.Delete(key)
			}()
			trace_log.DebugLogger(sn, "BinaryReq key:", key, " request timeout")
			return nil, noReplyErr
		} else {
			return l.TestRequest(sn, key, data)
		}
	case d := <-ch:
		defer func() {
			l.requestRetryMap.Delete(key)
		}()
		trace_log.DebugLogger(sn, "BinaryReq key:", key, " request success")
		return d, nil
	}
}

// SyncRequest 服务端向设备发送请求，并得到响应，有定时重试机制
func (l *session) SyncRequest(sn, requestKey string, pkt *model.Packet) (interface{}, error) {
	var (
		err  error
		ch   = make(chan interface{})
		data = pkt.ToHexBytes()
	)

	log.Infof("request %v %v", zap.String("sn", sn), zap.String("key", requestKey))
	l.requests.Store(requestKey, ch)
	defer func() {
		// delete request
		l.requests.Delete(requestKey)
	}()

	_, ok := l.requests.Load(requestKey)
	if ok {
		log.Info("store request ok", zap.String("key", requestKey))
	}

	if err = l.Send(sn, data); err != nil {
		log.Errorf("send data to device error %s", err)
		trace_log.DebugLogger(sn, "SyncRequest send data to device error", err)
		return nil, fmt.Errorf("发送数据失败")
	}

	// 记录重试次数
	retryCount, ok := l.requestRetryMap.Load(requestKey)
	if ok {
		retryCount = retryCount.(int) + 1
	} else {
		retryCount = 1
	}
	l.requestRetryMap.Store(requestKey, retryCount)

	log.Infof("wait for device [%s] response", sn)
	trace_log.CommandLogger(sn, fmt.Sprintf("SyncRequest send data to device data:[%X],", data))

	// 重复发，直到收到响应，或者超过重试次数
	t := time.Second * model.DefaultRequestTimeout
	timer := time.NewTimer(t)
	select {
	case <-timer.C:
		if retryCount.(int) >= requestRetryCount {
			// 达到重试次数
			defer l.requestRetryMap.Delete(requestKey)
			trace_log.CommandLogger(sn, fmt.Sprintf("SyncRequest send data to device[%v] after (%v,%v) ,failed data:[%X],err:%+v", sn, t.String(), requestRetryCount, data, noReplyErr))
			return nil, noReplyErr
		} else {
			// 重复发
			return l.SyncRequest(sn, requestKey, pkt)
		}
	case d := <-ch:
		defer func() {
			l.requestRetryMap.Delete(requestKey)
			if timer != nil {
				timer.Stop()
			}
		}()

		trace_log.CommandLogger(sn, fmt.Sprintf("SyncRequest send data to device[%v] success data:[%X],resp:+%X", sn, data, d))
		return d, nil
	}
}

// Reply 服务端向设备应答
func (l *session) Reply(requestKey string, requestPkt *model.Packet) {
	log.Infof("handle key %s response, data: %X", requestKey, requestPkt.Body)

	chRes, ok := l.requests.Load(requestKey)
	if !ok {
		log.Errorf("not found this request: %s", requestKey)
		return
	}
	l.requests.Delete(requestKey)

	if chRes != nil {
		ch := chRes.(chan interface{})
		ch <- requestPkt.Body
	}

	split := strings.Split(requestKey, "-")
	if err := l.ReleaseID(split[0], requestPkt.ID); err != nil {
		log.Error("request id release failed", err)
	}
	log.Debug("Response receive success")
}

// ReplyHeartBeat 向设备回复心跳
func (l *session) ReplyHeartBeat(sn, requestKey string, requestPkt *model.Packet) error {
	data := requestPkt.ToHexBytes()
	log.Infof("request %v %v", zap.String("sn", sn), zap.String("key", requestKey))
	// 心跳完，删除该记录
	defer l.requests.Delete(requestKey)

	_, ok := l.requests.Load(requestKey)
	if ok {
		log.Info("store request ok", zap.String("key", requestKey))
	}
	if err := l.Send(sn, data); err != nil {
		log.Errorf("send data to device error %s", err)
		trace_log.DebugLogger(sn, "SyncRequest send data to device error", err)
		// 记录重试次数
		retryCount, ok := l.requestRetryMap.Load(requestKey)
		if ok {
			retryCount = retryCount.(int) + 1
		} else {
			retryCount = 1
		}
		if retryCount.(int) >= replyHeartBeatCount {
			return heartbeatErr
		}
		l.requestRetryMap.Store(requestKey, retryCount)
		return l.ReplyHeartBeat(sn, requestKey, requestPkt)
	}
	return nil
}

func (l *session) Disconnect(c gnet.Conn, sn string, callback func(sn string) error) {
	defer utils.ExceptionCatch()
	var (
		ch chan string
	)

	trace_log.DebugLogger(sn, "开始断开连接 Disconnect close connId:", c.Fd())
	if callback != nil {
		ch = make(chan string, 1)
		l.CloseMap.Store(sn, ch)
	}

	if c != nil {
		if err := c.Close(); err != nil {
			trace_log.DebugLogger(sn, "Disconnect close conn error", err)
		}
	}

	if callback != nil {
		for {
			select {
			case data := <-ch:
				trace_log.DebugLogger(sn, "Disconnect 断开连接 connId:", c.Fd())
				l.CloseMap.Delete(sn)
				_ = callback(data)
				return
			case <-time.After(time.Second * 20):
				trace_log.DebugLogger(sn, "Disconnect 超时断开连接 connId:", c.Fd())
				l.CloseMap.Delete(sn)
				_ = callback(sn)
				return
			}
		}
	}
}

func (l *session) ClearConn(connId int, sn string) {

	l.status.Delete(connId)
	res, ok := l.conns.Load(sn)
	if ok && res != nil {
		l.conns.Delete(sn)
	}

	l.cachePacket.Delete(connId)

	l.requests.Range(func(k, v any) bool {
		key := k.(string)
		if strings.Contains(key, sn) {
			l.requests.Delete(key)
		}
		return true
	})

	trace_log.DebugLogger(sn, "ClearConn session Disconnect connId:", connId)
}

func (l *session) checkCloseChan(closeChan chan *closeData) {
	defer utils.ExceptionCatch()
	for closeData := range closeChan {
		if v, ok := l.CloseMap.Load(closeData.Sn); ok {
			ch := v.(chan string)
			if ch == nil {
				continue
			}

			trace_log.DebugLogger(closeData.Sn, "checkCloseChan 检查到关闭消息:")
			ch <- closeData.Sn
		}
	}
}

func (l *session) checkConnections(ch chan struct{}) {
	defer utils.ExceptionCatch()
	ticker := time.NewTicker(time.Second * 10)
	for {
		select {
		case <-ch:
			return
		case <-ticker.C:
			length := 0
			l.status.Range(func(k, v any) bool {
				key := k.(int)
				status := v.(*status)

				if status.keepalive <= time.Now().Unix() && status.IsExpire == false {
					status.IsExpire = true
					trace_log.CommonLogger.Info("---------- tcp 心跳检查超时 session checkConnections sn: ", string(status.sn), " ip:", key)
					trace_log.DebugLogger(string(status.sn), " tcp 心跳检查超时 session checkConnections ip", key)
					if l.conns != nil {
						if c, ok := l.conns.Load(string(status.sn)); ok {
							goroutine_pool.GetPoolV2("close").Push(c, func(data interface{}) error {
								if data.(gnet.Conn) == nil {
									trace_log.DebugLogger(string(status.sn), "conn is nil")
									return nil
								}
								l.OnClose(data.(gnet.Conn), nil)
								_ = data.(gnet.Conn).Close()
								return nil
							})
						}
					}
				} else {
					length++
				}
				return true
			})

			// 打印当前连接数
			fmt.Println("############# 当前连接数:", length)
			fmt.Println("############# 当前连接数:", length)
			fmt.Println("############# 当前连接数:", length)
			fmt.Println("############# 当前连接数:", length)
			fmt.Println("############# 当前连接数:", length)
		}
	}
}

// OnOpen 有新连接建立
func (l *session) OnOpen(c gnet.Conn) {
	atomic.AddInt32(&l.connectCount, 1)
	trace_log.CommonLogger.Info("OnOpen 建立了一个新连接 got a connection connId:", c.Fd(), " connectCount:", l.connectCount)
}

// OnClose 有连接关闭
func (l *session) OnClose(c gnet.Conn, err error) {
	defer utils.ExceptionCatch()
	trace_log.CommonLogger.Info("connId:", c.Fd(), " OnClose 2 关闭了一个新连接")
	sn := Session.GetSnByConnId(c.Fd())
	if sn == "" {
		trace_log.ErrorLogger.Error("OnClose 获取不到设备号 connId:", c.Fd())
		return
	}
	trace_log.DebugLogger(sn, "OnClose tcp 连接被断开 connection closed connId:", c.Fd())

	// 清除连接
	Session.ClearConn(c.Fd(), sn)

	// 计数
	atomic.AddInt32(&l.connectCount, -1)
	trace_log.CommonLogger.Info("OnClose 设备端断开了一个连接 connId:", c.Fd(), " connectCount:", l.connectCount)

	// 设备下线，保存状态
	_ = Device.Offline(sn)

	// 设备离线
	eData := &dto.DeviceStatus{
		Sn:    sn,
		Event: dto.EventType_Offline,
	}
	event.E.PublishSync(event.DeviceStatusChange, eData)

	// 没有信号
	ed := &dto.Signal{
		Sn:        sn,
		SignalGps: int32(0),
		Signal_4G: int32(0),
	}

	event.E.PublishSync(event.DeviceSignalReport, ed)

	trace_log.DebugLogger(sn, "OnClose 发送关闭消息 connId:", c.Fd(), " sn:", sn, " err:", err)
	// 关闭
	Session.CloseChan <- &closeData{
		Sn:     sn,
		ConnId: c.Fd(),
	}
}

func (l *session) GetStatus(connId int) *status {
	res, ok := l.status.Load(connId)
	if ok && res != nil {
		return res.(*status)
	}
	return nil
}

func (l *session) LoadCachePacket(fd int) []byte {
	res, ok := l.cachePacket.Load(fd)
	if ok {
		return res.([]byte)
	}
	return nil
}

func (l *session) ResetCachePacket(fd int) {
	res, ok := l.cachePacket.Load(fd)
	if ok {
		res = res.([]byte)[:0]
	}

	return
}

func (l *session) AppendCachePacket(fd int, b []byte, before bool) {
	res, ok := l.cachePacket.Load(fd)
	if ok {
		if before {
			res = append(res.([]byte), b...)
		} else {
			res = append(b, res.([]byte)...)
		}
	}

	return
}

func (l *session) SaveCachePacket(fd int, b []byte) {
	l.cachePacket.Store(fd, b)
}
